package com.hmall.transport.netty.codec;

import com.hmall.compress.Compress;
import com.hmall.compress.gzip.GzipCompress;
import com.hmall.enums.CompressTypeEnum;
import com.hmall.enums.SerializationTypeEnum;
import com.hmall.extension.ExtensionLoader;
import com.hmall.serialize.Kryo.KryoSerializer;
import com.hmall.remoting.constants.RpcContants;
import com.hmall.remoting.dto.RpcMessage;
import com.hmall.remoting.dto.RpcRequest;
import com.hmall.remoting.dto.RpcResponse;
import com.hmall.serialize.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;

/**
 * 自定义协议解码器
 * <pre>
 *   0     1     2     3     4        5     6     7     8         9          10      11     12  13  14   15 16
 *   +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+
 *   |   magic   code        |version | full length         | messageType| codec|compress|    RequestId       |
 *   +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+
 *   |                                                                                                       |
 *   |                                         body                                                          |
 *   |                                                                                                       |
 *   |                                        ... ...                                                        |
 *   +-------------------------------------------------------------------------------------------------------+
 * 4B  magic code（魔法数）   1B version（版本）   4B full length（消息长度）    1B messageType（消息类型）
 * 1B compress（压缩类型） 1B codec（序列化类型）    4B  requestId（请求的Id）
 * body（object类型数据）
 * </pre>
 * <p>
 * {@link LengthFieldBasedFrameDecoder} 是一种基于长度的解码器，用于解决TCP解包和粘滞问题。
 * </p>
 *
 * @author wangtao
 * @createTime on 2020/10/2
 * @see <a href="https://zhuanlan.zhihu.com/p/95621344">LengthFieldBasedFrameDecoder解码器</a>
 */
@Slf4j
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {

    public RpcMessageDecoder() {
        this(RpcContants.MAX_FRAME_LENGTH,5,4,-9,0);
    }

    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decode = super.decode(ctx, in);//尝试从缓存区获取一条完整的信息，若数据不全返回null
        if (decode instanceof ByteBuf) {
            ByteBuf buf = (ByteBuf) decode;
            if (buf.readableBytes() >= RpcContants.TOTAL_LENGTH) {//判断可读数据是否大于一条完整数据的最小长度
                try {
                    return decodeFrame(buf);
                } catch (Exception e) {
                    log.error("Decode frame error!", e);
                    throw new RuntimeException(e);
                }
            }
        }
        return decode;
    }

    private Object decodeFrame(ByteBuf buf) {
        checkMagicNumber(buf);
        checkVersion(buf);
        int fullLength = buf.readInt();
        byte messageType = buf.readByte();
        byte codec = buf.readByte();
        byte compressType = buf.readByte();
        int requestId = buf.readInt();
        RpcMessage rpcMessage = RpcMessage.builder()
                .messageType(messageType)
                .compress(compressType)
                .codec(codec)
                .requestId(requestId)
                .build();
        if (messageType == RpcContants.HEARTBEAT_REQUEST_TYPE){
            rpcMessage.setData(RpcContants.PING);
            return rpcMessage;
        }
        if (messageType == RpcContants.HEARTBEAT_RESPONSE_TYPE){
            rpcMessage.setData(RpcContants.PONG);
        }
        int bodyLength = fullLength -RpcContants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] body = new byte[bodyLength];
            buf.readBytes(body);
            // decompress the bytes
            String compressName = CompressTypeEnum.getClazz(compressType);
            Compress compress = ExtensionLoader.getExtensionLoader(Compress.class)
                    .getExtension(compressName);
            body = compress.decompress(body);
            // deserialize the object
            String codecName = SerializationTypeEnum.getClazz(rpcMessage.getCodec());
            log.info("codec name: [{}] ", codecName);
            Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class)
                    .getExtension(codecName);
            if (messageType == RpcContants.REQUEST_TYPE){
                RpcRequest tmpValue = serializer.deserialize(body, RpcRequest.class);//反序列化
                rpcMessage.setData(tmpValue);
            }else {
                RpcResponse tmpValue = serializer.deserialize(body, RpcResponse.class);
                rpcMessage.setData(tmpValue);
            }
        }
        return rpcMessage;
    }

    private void checkVersion(ByteBuf buf) {
        byte version = buf.readByte();
        if (version != RpcContants.VERSION) {
            throw new RuntimeException("version isn't compatible" + version);
        }
    }

    private void checkMagicNumber(ByteBuf buf) {
        byte[] tmp = new byte[RpcContants.MAGIC_NUMBER.length];
        buf.readBytes(tmp);
        for (int i = 0; i < tmp.length; i++) {
            if (tmp[i] != RpcContants.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("Unknown magic code: " + Arrays.toString(tmp));
            }
        }
    }
}
